package com.baishancloud.log.douyin.demand1

import com.baishancloud.log.common.entity.ServerCacheGroupEntity
import com.baishancloud.log.common.util.{ExternalDataFetchUtil, IpInfoUtil, TimeUtil}
import com.baishancloud.log.douyin.mallardName1
import com.baishancloud.log.format.scala.Origin
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

import scala.collection.mutable

/**
 *
 * @author ziqiang.wang 
 * @date 2022/3/1 16:01
 */
class LogParse1 extends BroadcastProcessFunction[Origin, mutable.Map[String, ServerCacheGroupEntity], Result1] {

  private val logger: Logger = LoggerFactory.getLogger(classOf[LogParse1])

  private var serverCacheGroupInfo: mutable.Map[String, ServerCacheGroupEntity] = mutable.Map[String, ServerCacheGroupEntity]()

  override def open(parameters: Configuration): Unit = {
    //获取外部资源
    ExternalDataFetchUtil.serverCacheGroupFetch(serverCacheGroupInfo)
  }

  override def processElement(value: Origin, ctx: BroadcastProcessFunction[Origin, mutable.Map[String, ServerCacheGroupEntity], Result1]#ReadOnlyContext, out: Collector[Result1]): Unit = {
    out.collect(
      Result1(
        TimeUtil.timeAtLong(value.datetimeOfUTC().toEpochSecond, 60),
        60,
        mallardName1,
        "",
        1L,
        Tags(
          value.domain(),
          IpInfoUtil.core(value.ip()).isp,
          serverCacheGroupInfo.getOrElse(value.host(), ServerCacheGroupEntity.DEFAULT).cacheGroupCode
        ),
        Fields(
          if (!value.complexField0().split("@")(3).contains("dx-lt")) 1L else 0L
        )
      )
    )
  }

  override def processBroadcastElement(value: mutable.Map[String, ServerCacheGroupEntity], ctx: BroadcastProcessFunction[Origin, mutable.Map[String, ServerCacheGroupEntity], Result1]#Context, out: Collector[Result1]): Unit = {
    if (value.nonEmpty) {
      serverCacheGroupInfo = value
      logger.info(s"接收到广播数据，ipCacheGroupInfo表size为：${serverCacheGroupInfo.size}")
    }
  }
}
